package com.streamsets.datacollector.pileline.stage;

import com.streamsets.datacollector.model.etl.FieldValueReplacerConfigJson;
import com.streamsets.datacollector.pileline.enums.CompareConditionEnum;
import com.streamsets.datacollector.pileline.enums.FieldValueReplacerConfigEnum;
import com.streamsets.datacollector.restapi.bean.ConfigConfigurationJson;
import com.streamsets.datacollector.restapi.bean.StageConfigurationJson;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 *  初始化 管道 值替换Stage的配置
 *  stage的完整描述示例如：
 *  {
 *       "instanceName" : "g1_01",
 *       "library" : "streamsets-datacollector-basic-lib",
 *       "stageName" : "com_streamsets_pipeline_stage_processor_fieldvaluereplacer_FieldValueReplacerDProcessor",
 *       "stageVersion" : "3",
 *       "configuration" : [ {
 *         "name" : "nullReplacerConditionalConfigs",
 *         "value" : [ {
 *           "fieldsToNull" : [ "/created_by" ]
 *         }, {
 *           "fieldsToNull" : [ "/updated_dt" ]
 *         } ]
 *       }, {
 *         "name" : "fieldsToReplaceIfNull",
 *         "value" : [ {
 *           "fields" : [ "/remark" ],
 *           "newValue" : "备注"
 *         }, {
 *           "fields" : [ "/status" ],
 *           "newValue" : "2"
 *         } ]
 *       }, {
 *         "name" : "onStagePreConditionFailure",
 *         "value" : "TO_ERROR"
 *       }, {
 *         "name" : "fieldsToConditionallyReplace",
 *         "value" : [ {
 *           "fieldNames" : [ "/role_type" ],
 *           "operator" : "LESS_THAN",
 *           "comparisonValue" : "0",
 *           "replacementValue" : "通用角色"
 *         }, {
 *           "fieldNames" : [ "/role_type" ],
 *           "operator" : "EQUALS",
 *           "comparisonValue" : "1",
 *           "replacementValue" : "机构角色"
 *         }, {
 *           "fieldNames" : [ "/sort_no" ],
 *           "operator" : "GREATER_THAN",
 *           "comparisonValue" : "1",
 *           "replacementValue" : "22"
 *         } ]
 *       }, {
 *         "name" : "stageOnRecordError",
 *         "value" : "TO_ERROR"
 *       }, {
 *         "name" : "stageRequiredFields",
 *         "value" : [ ]
 *       }, {
 *         "name" : "stageRecordPreconditions",
 *         "value" : [ ]
 *       } ],
 *       "uiInfo" : {
 *         "description" : "",
 *         "label" : "值转换",
 *         "xPos" : -558.7707750769339,
 *         "yPos" : -49.12270880114836,
 *         "stageType" : "PROCESSOR"
 *       },
 *       "inputLanes" : [ "HiveMetadata_01OutputLane15857327993100" ],
 *       "outputLanes" : [ "g1_01OutputLane15857262190730" ],
 *       "eventLanes" : [ ],
 *       "services" : [ ]
 *     }
 *
 * @author wengzhonghui
 * @date 2020/4/2 10:32
 */
public class InitFieldValueReplacerConfig {

    /*
     * stage instanceName名称
     */
    public static final String STAGE_NAME = "g1_01";

    public static boolean init(StageConfigurationJson stage, FieldValueReplacerConfigJson configDto){
        if(STAGE_NAME.equalsIgnoreCase(stage.getInstanceName())) {
            if (configDto != null) {
                List<ConfigConfigurationJson> configs = intConfig(stage.getConfiguration(), configDto);
                stage.setConfiguration(configs);
            } else {
                List<ConfigConfigurationJson> configs =
                    stage.getConfiguration().stream().filter(e -> "nullReplacerConditionalConfigs".equals(e.getName())
                        || "fieldsToReplaceIfNull".equals(e.getName())|| "fieldsToConditionallyReplace".equals(e.getName())
                    ).collect(Collectors
                        .toList());
                stage.getConfiguration().removeAll(configs);
            }
        }
        return true;
    }

    public static List<ConfigConfigurationJson> intConfig(List<ConfigConfigurationJson> oldConfigs, FieldValueReplacerConfigJson configDto){
        List<ConfigConfigurationJson> newConfigs = new ArrayList<>();
        for(ConfigConfigurationJson config :oldConfigs ){
            config = initNullReplacerConditionalConfigs(config, configDto.getNullReplaceColumns());
            config = initFieldsToReplaceIfNullConfigs(config, configDto.getFieldsToReplaceIfNullConfigs());
            config = initFieldsToConditionallyReplaceConfigs(config, configDto.getFieldsToConditionallyReplaceConfigs());
            newConfigs.add(config);
        }
        return newConfigs;
    }

    /*
     * 把 列值替换为null的配置
     * 格式示例如：
     * [ {
          "fieldsToNull" : [ "/created_by" ]
        }, {
          "fieldsToNull" : [ "/updated_dt" ]
        } ]
     * @param [config 原来的配置, columnNames:需要替换为null的列名列表]
     * @return 返回新生成的配置
     */
    private static ConfigConfigurationJson initNullReplacerConditionalConfigs(ConfigConfigurationJson config, List<String> columnNames){
        if(FieldValueReplacerConfigEnum.nullReplacerConditionalConfigs.getName().equalsIgnoreCase(config.getName())){
            List configValList = new ArrayList();
            if(columnNames != null && columnNames.size() > 0){
                for(String name : columnNames){
                    Map<String,String[]> data = new HashMap();
                    String[] val = { "/" + name };
                    data.put("fieldsToNull", val);
                    configValList.add(data);
                }
            }
            config = new ConfigConfigurationJson(FieldValueReplacerConfigEnum.nullReplacerConditionalConfigs.getName()
                    ,configValList);
        }
        return config;
    }


    /*
     * 把当值为空时，替换为某个值
     * 格式示例如：
     * [ {
          "fields" : [ "/remark" ],
          "newValue" : "备注"
        }, {
          "fields" : [ "/status" ],
          "newValue" : "2"
        } ]
     * @param [config 原来的配置, columnNames:需要替换为null的列名列表]
     * @return 返回新生成的配置
     */
    private static ConfigConfigurationJson initFieldsToReplaceIfNullConfigs(ConfigConfigurationJson config, List<Map<String, String>> fieldsToReplaceIfNullConfigs){
        if(FieldValueReplacerConfigEnum.fieldsToReplaceIfNull.getName().equalsIgnoreCase(config.getName())){
            List configValList = new ArrayList();
            if(fieldsToReplaceIfNullConfigs != null && fieldsToReplaceIfNullConfigs.size() > 0){
                for(Map<String, String> valConfig : fieldsToReplaceIfNullConfigs){
                    Map data = new HashMap();
                    String[] val = { "/" + valConfig.get("fields") };
                    data.put("fields", val);
                    data.put("newValue", valConfig.get("newValue"));
                    configValList.add(data);
                }
            }
            config = new ConfigConfigurationJson(FieldValueReplacerConfigEnum.fieldsToReplaceIfNull.getName()
                    ,configValList);
        }
        return config;
    }

    /*
     * 把当值为空时，替换为某个值
     * 格式示例如：
     * [ {
          "fieldNames" : [ "/role_type" ],
          "operator" : "LESS_THAN",
          "comparisonValue" : "0",
          "replacementValue" : "通用角色"
        }, {
          "fieldNames" : [ "/role_type" ],
          "operator" : "EQUALS",
          "comparisonValue" : "1",
          "replacementValue" : "机构角色"
        }, {
          "fieldNames" : [ "/sort_no" ],
          "operator" : "GREATER_THAN",
          "comparisonValue" : "1",
          "replacementValue" : "22"
        } ]
     * @param [config 原来的配置, columnNames:需要替换为null的列名列表]
     * @return 返回新生成的配置
     */
    private static ConfigConfigurationJson initFieldsToConditionallyReplaceConfigs(ConfigConfigurationJson config, List<Map<String, String>> fieldsToConditionallyReplaceConfigs){
        if(FieldValueReplacerConfigEnum.fieldsToConditionallyReplace.getName().equalsIgnoreCase(config.getName())){
            List configValList = new ArrayList();
            if(fieldsToConditionallyReplaceConfigs != null && fieldsToConditionallyReplaceConfigs.size() > 0){
                for(Map<String, String> valConfig : fieldsToConditionallyReplaceConfigs){
                    Map data = new HashMap();
                    String fieldName = valConfig.get("fields");
                    String operator =  valConfig.get("operator");
                    String comparisonValue =  valConfig.get("comparisonValue");
                    String replacementValue =  valConfig.get("replacementValue");
                    if(StringUtils.isEmpty(fieldName) || StringUtils.isEmpty(operator)){
                        continue;
                    }
                    comparisonValue = comparisonValue == null ? "0" : comparisonValue;
                    replacementValue = replacementValue == null ? "0" : replacementValue;
                    operator = operator.trim();
                    String operatorDesc = CompareConditionEnum.defaultValueOfName(operator);

                    String[] fieldNames = { "/" + fieldName };
                    data.put("fieldNames", fieldNames);
                    data.put("operator", operatorDesc);
                    data.put("comparisonValue", comparisonValue);
                    data.put("replacementValue", replacementValue);

                    configValList.add(data);
                }
            }
            config = new ConfigConfigurationJson(FieldValueReplacerConfigEnum.fieldsToConditionallyReplace.getName()
                    ,configValList);
        }
        return config;
    }

}
